/*
* Created by boil on 2023/10/26.
*/

#include "tcp_server.h"
#include "utils/once_token.h"
#include "net/event_loop_pool.h"
#include "utils/string_util.h"

RD_NAMESPACE_BEGIN

  TcpServer::TcpServer(const EventLoop::Ptr &poller) : AService(poller) {
    setOnCreateSocket(nullptr);
  }

  void TcpServer::setupEvent() {
    _socket = createSocket(_poller);
    std::weak_ptr<TcpServer> weak_self = std::static_pointer_cast<TcpServer>(shared_from_this());
    _socket->setOnBeforeAccept([weak_self](const EventLoop::Ptr &poller) -> Socket::Ptr {
      if (auto strong_self = weak_self.lock()) {
        return strong_self->onBeforeAcceptConnection(poller);
      }
      return nullptr;
    });
    _socket->setOnAccept([weak_self](Socket::Ptr &sock, std::shared_ptr<void> &complete) {
      if (auto strong_self = weak_self.lock()) {
        auto ptr = sock->getPoller().get();
        auto server = strong_self->getServer(ptr);
        ptr->Async([server, sock, complete]() {
          //该tcp客户端派发给对应线程的TcpServer服务器
          server->onAcceptConnection(sock);
        });
      }
    });
  }

  TcpServer::~TcpServer() {
    if (_main_server && _socket && _socket->rawFD() != -1) {
      LOG_INFO << "Close tcp server [" << _socket->GetLocalIp() << "]: " << _socket->GetLocalPort();
    }
    _timer.reset();
    //先关闭socket监听，防止收到新的连接
    _socket.reset();
    _session_map.clear();
    _cloned_server.clear();
  }

  uint16_t TcpServer::getPort() {
    if (!_socket) {
      return 0;
    }
    return _socket->GetLocalPort();
  }

  void TcpServer::setOnCreateSocket(Socket::onCreateSocket cb) {
    if (cb) {
      _on_create_socket = std::move(cb);
    } else {
      _on_create_socket = [](const EventLoop::Ptr &poller) {
        return Socket::CreateSocket(poller, false);
      };
    }
    for (auto &pr : _cloned_server) {
      pr.second->setOnCreateSocket(cb);
    }
  }

  TcpServer::Ptr TcpServer::onCreatServer(const EventLoop::Ptr &poller) {
    return std::make_shared<TcpServer>(poller);
  }

  Socket::Ptr TcpServer::onBeforeAcceptConnection(const EventLoop::Ptr &poller) {
    assert(_poller->isCurrentThread());
    //此处改成自定义获取poller对象，防止负载不均衡
    return createSocket(EventLoopPool::Instance().getPoller(false));
  }

  void TcpServer::cloneFrom(const TcpServer &that) {
    if (!that._socket) {
      throw std::invalid_argument("TcpServer::cloneFrom other with null socket");
    }
    setupEvent();
    _main_server = false;
    _on_create_socket = that._on_create_socket;
    _session_alloc = that._session_alloc;
    std::weak_ptr<TcpServer> weak_self = std::static_pointer_cast<TcpServer>(shared_from_this());
    _timer = std::make_shared<Timer>(2.0f, [weak_self]() -> bool {
      auto strong_self = weak_self.lock();
      if (!strong_self) {
        return false;
      }
      strong_self->onManagerSession();
      return true;
    }, _poller);
//    this->mINI::operator=(that);
    _parent = static_pointer_cast<TcpServer>(const_cast<TcpServer &>(that).shared_from_this());
    _socket->cloneSocket(*(that._socket));
  }

// 接收到客户端连接请求
  Session::Ptr TcpServer::onAcceptConnection(const Socket::Ptr &sock) {
    assert(_poller->isCurrentThread());
    std::weak_ptr<TcpServer> weak_self = std::static_pointer_cast<TcpServer>(shared_from_this());
    //创建一个Session;这里实现创建不同的服务会话实例
    auto helper = _session_alloc(std::static_pointer_cast<TcpServer>(shared_from_this()), sock);
    auto session = helper->session();
    //把本服务器的配置传递给Session
    session->attachServer(*this);

    //_session_map::emplace肯定能成功
    auto success = _session_map.emplace(helper.get(), helper).second;
    assert(success == true);

    std::weak_ptr<Session> weak_session = session;
    //会话接收数据事件
    sock->setOnRead([weak_session](const ABuffer::Ptr &buf, struct sockaddr *, int) {
      //获取会话强应用
      auto strong_session = weak_session.lock();
      if (!strong_session) {
        return;
      }
      try {
        strong_session->onRecv(buf);
      } catch (SockException &ex) {
        strong_session->Shutdown(ex);
      } catch (std::exception &ex) {
        strong_session->Shutdown(SockException(Err_shutdown, ex.what()));
      }
    });

    SessionHelper *ptr = helper.get();
    auto cls = ptr->className();
    //会话接收到错误事件
    sock->setOnErr([weak_self, weak_session, ptr, cls](const SockException &err) {
      //在本函数作用域结束时移除会话对象
      //目的是确保移除会话前执行其onError函数
      //同时避免其onError函数抛异常时没有移除会话对象
      OnceToken token(nullptr, [&]() {
        //移除掉会话
        auto strong_self = weak_self.lock();
        if (!strong_self) {
          return;
        }

        assert(strong_self->_poller->isCurrentThread());
        if (!strong_self->_is_on_manager) {
          //该事件不是onManager时触发的，直接操作map
          strong_self->_session_map.erase(ptr);
        } else {
          //遍历map时不能直接删除元素
          strong_self->_poller->Async([weak_self, ptr]() {
            auto strong_self = weak_self.lock();
            if (strong_self) {
              strong_self->_session_map.erase(ptr);
            }
          }, false);
        }
      });

      //获取会话强应用
      auto strong_session = weak_session.lock();
      if (strong_session) {
        //触发onError事件回调
        TraceP(strong_session) << cls << " on err: " << err;
        strong_session->onError(err);
      }
    });
    return session;
  }

  void TcpServer::listen(uint16_t port, const std::string &host, uint32_t backlog) {
    setupEvent();
    if (!_socket->listen(port, host.c_str(), backlog)) {
      //创建tcp监听失败，可能是由于端口占用或权限问题
      string err = (StrPrinter << "Listen on " << host << " " << port << " failed: " << GetErrorMsg(true));
      throw std::runtime_error(err);
    }

    //新建一个定时器定时管理这些tcp会话
    std::weak_ptr<TcpServer> weak_self = std::static_pointer_cast<TcpServer>(shared_from_this());
    _timer = std::make_shared<Timer>(2.0f,[weak_self]() -> bool {
      auto strong_self = weak_self.lock();
      if (!strong_self) {
        return false;
      }
      strong_self->onManagerSession();
      return true;
    },  _poller);

    EventLoopPool::Instance().for_each([&](const TaskExecutor::Ptr &executor) {
      EventLoop::Ptr poller = static_pointer_cast<EventLoop>(executor);
      if (poller == _poller) {
        return;
      }
      auto &serverRef = _cloned_server[poller.get()];
      if (!serverRef) {
        serverRef = onCreatServer(poller);
      }
      if (serverRef) {
        serverRef->cloneFrom(*this);
      }
    });

    LOG_INFO << "TCP server listening on [" << host << "]: " << port;
  }

  void TcpServer::onManagerSession() {
    assert(_poller->isCurrentThread());

    OnceToken token([&]() {
      _is_on_manager = true;
    }, [&]() {
      _is_on_manager = false;
    });

    for (auto &pr : _session_map) {
      //遍历时，可能触发onErr事件(也会操作_session_map)
      try {
        pr.second->session()->onManager();
      } catch (std::exception &ex) {
        LOG_WARN << ex.what();
      }
    }
  }

  Socket::Ptr TcpServer::createSocket(const EventLoop::Ptr &poller) {
    return _on_create_socket(poller);
  }

  TcpServer::Ptr TcpServer::getServer(const EventLoop *poller) const {
    auto parent = _parent.lock();
    auto &ref = parent ? parent->_cloned_server : _cloned_server;
    auto it = ref.find(poller);
    if (it != ref.end()) {
      //派发到cloned server
      return it->second;
    }
    //派发到parent server
    return static_pointer_cast<TcpServer>(parent ? parent : const_cast<TcpServer *>(this)->shared_from_this());
  }

  Session::Ptr TcpServer::createSession(const Socket::Ptr &sock) {
    return getServer(sock->getPoller().get())->onAcceptConnection(sock);
  }

RD_NAMESPACE_END


